package com.learn.join;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichCoGroupFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;

/**
 * @create: 2023-04-24 15:15
 * @author: Mr.Du
 * --------------
 * @notes:
 **/
public class TumblingWindowCoGroupDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        //测试的时候设置并行度为1，因为窗口计算的触发要等所有分区的窗口结束
        //另外窗口join是整个窗口的join，不是分区join，整个要理解
        env.setParallelism(1);
        DataStreamSource<String> mainOrderSource = env.socketTextStream("node1", 9999);
        DataStreamSource<String> detailOrderSource = env.socketTextStream("node1", 9998);

        SingleOutputStreamOperator<MainOrder> mainOrderStringKeyedStream = mainOrderSource.map(new MapFunction<String, MainOrder>() {
            @Override
            public MainOrder map(String s) throws Exception {
                String[] arr = s.split(",");
                return new MainOrder(arr[0], arr[1], arr[2], arr[3], Double.parseDouble(arr[4]));
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<MainOrder>forBoundedOutOfOrderness(Duration.ZERO)
                .withTimestampAssigner(new SerializableTimestampAssigner<MainOrder>() {
                    @Override
                    public long extractTimestamp(MainOrder mainOrder, long l) {
                        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        try {
                            return format.parse(mainOrder.getOrderTime()).getTime();
                        } catch (ParseException e) {
                            e.printStackTrace();
                            return 0L;
                        }
                    }
                }));//.keyBy(MainOrder::getOrderId);

        mainOrderStringKeyedStream.map(new RichMapFunction<MainOrder, String>() {
            @Override
            public String map(MainOrder mainOrder) throws Exception {
                return getRuntimeContext().getIndexOfThisSubtask() + ":" + mainOrder;
            }
        }).print();


        SingleOutputStreamOperator<DetailOrder> detailOrderStringKeyedStream = detailOrderSource.map(new MapFunction<String, DetailOrder>() {
            @Override
            public DetailOrder map(String s) throws Exception {
                String[] arr = s.split(",");
                return new DetailOrder(arr[0], arr[1], arr[2], arr[3], Integer.parseInt(arr[4]), Double.parseDouble(arr[5]));
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<DetailOrder>forBoundedOutOfOrderness(Duration.ZERO)
                .withTimestampAssigner(new SerializableTimestampAssigner<DetailOrder>() {
                    @Override
                    public long extractTimestamp(DetailOrder detailOrder, long l) {
                        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        try {
                            return format.parse(detailOrder.getDetailTime()).getTime();
                        } catch (ParseException e) {
                            e.printStackTrace();
                            return 0L;
                        }
                    }
                }));//.keyBy(DetailOrder::getOrderId);

        detailOrderStringKeyedStream.map(new RichMapFunction<DetailOrder, String>() {
            @Override
            public String map(DetailOrder detailOrder) throws Exception {
                return getRuntimeContext().getIndexOfThisSubtask() + ": " + detailOrder;
            }
        }).print();
        mainOrderStringKeyedStream.coGroup(detailOrderStringKeyedStream).where(new KeySelector<MainOrder, String>() {
            @Override
            public String getKey(MainOrder mainOrder) throws Exception {
                return mainOrder.getOrderId();
            }
        }).equalTo(DetailOrder::getOrderId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .apply(new RichCoGroupFunction<MainOrder, DetailOrder, DwdOrder>() {
                    @Override
                    public void coGroup(Iterable<MainOrder> mainOrderIterable, Iterable<DetailOrder> detailOrderIterable, Collector<DwdOrder> out) throws Exception {

                        int taskId = getRuntimeContext().getIndexOfThisSubtask();
                        System.out.println("current taskId is: " + taskId);
                        for (MainOrder mainOrder : mainOrderIterable) {
                            System.out.println(mainOrder);
                        }
                        for (DetailOrder detailOrder : detailOrderIterable) {
                            System.out.println(detailOrder);
                        }
                        //窗口内，同一个key的两个流的数据都在这里了，怎么join就看自己需求了
                        //以left join为例
                        for (MainOrder mainOrder : mainOrderIterable) {
                            String orderId = mainOrder.getOrderId();
                            boolean flag = false;
                            for (DetailOrder detailOrder : detailOrderIterable) {
                                if(detailOrder.getOrderId().equals(orderId)) {
                                    flag = true;
                                    DwdOrder dwdOrder = new DwdOrder();
                                    dwdOrder.setOrderTime(mainOrder.getOrderTime());
                                    dwdOrder.setOrderId(mainOrder.getOrderId());
                                    dwdOrder.setUserId(mainOrder.getUserId());
                                    dwdOrder.setAddress(mainOrder.getAddress());
                                    dwdOrder.setOrderMoney(mainOrder.getOrderMoney());
                                    dwdOrder.setDetailOrderTime(detailOrder.getDetailTime());
                                    dwdOrder.setDetailId(detailOrder.getDetailId());
                                    dwdOrder.setGoodsName(detailOrder.getGoodsName());
                                    dwdOrder.setGoodNumber(detailOrder.getGoodsNumber());
                                    dwdOrder.setDetailMoney(detailOrder.getDetailMoney());
                                    out.collect(dwdOrder);
                                }
                            }
                            if(!flag) {
                                DwdOrder dwdOrder = new DwdOrder();
                                dwdOrder.setOrderTime(mainOrder.getOrderTime());
                                dwdOrder.setOrderId(mainOrder.getOrderId());
                                dwdOrder.setUserId(mainOrder.getUserId());
                                dwdOrder.setAddress(mainOrder.getAddress());
                                dwdOrder.setOrderMoney(mainOrder.getOrderMoney());
                                out.collect(dwdOrder);
                            }
                        }

                    }
                }).print();
        /**mainOrderSingleOutputStreamOperator.coGroup(detailOrderSingleOutputStreamOperator)
                .where(new KeySelector<MainOrder, String>() {
                    @Override
                    public String getKey(MainOrder mainOrder) throws Exception {
                        return mainOrder.getOrderId();
                    }
                })
                .equalTo(DetailOrder::getOrderId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .apply(new CoGroupFunction<MainOrder, DetailOrder, DwdOrder>() {
                    @Override
                    public void coGroup(Iterable<MainOrder> mainOrderIterable, Iterable<DetailOrder> detailOrderIterable, Collector<DwdOrder> out) throws Exception {
                        //窗口内，同一个key的两个流的数据都在这里了，怎么join就看自己需求了
                        //以left join为例
                        for (MainOrder mainOrder : mainOrderIterable) {
                            String orderId = mainOrder.getOrderId();
                            boolean flag = false;
                            for (DetailOrder detailOrder : detailOrderIterable) {
                                if(detailOrder.getOrderId().equals(orderId)) {
                                    flag = true;
                                    DwdOrder dwdOrder = new DwdOrder();
                                    dwdOrder.setOrderTime(mainOrder.getOrderTime());
                                    dwdOrder.setOrderId(mainOrder.getOrderId());
                                    dwdOrder.setUserId(mainOrder.getUserId());
                                    dwdOrder.setAddress(mainOrder.getAddress());
                                    dwdOrder.setOrderMoney(mainOrder.getOrderMoney());
                                    dwdOrder.setDetailOrderTime(detailOrder.getDetailTime());
                                    dwdOrder.setDetailId(detailOrder.getDetailId());
                                    dwdOrder.setGoodsName(detailOrder.getGoodsName());
                                    dwdOrder.setGoodNumber(detailOrder.getGoodsNumber());
                                    dwdOrder.setDetailMoney(detailOrder.getDetailMoney());
                                    out.collect(dwdOrder);
                                }
                            }
                            if(!flag) {
                                DwdOrder dwdOrder = new DwdOrder();
                                dwdOrder.setOrderTime(mainOrder.getOrderTime());
                                dwdOrder.setOrderId(mainOrder.getOrderId());
                                dwdOrder.setUserId(mainOrder.getUserId());
                                dwdOrder.setAddress(mainOrder.getAddress());
                                dwdOrder.setOrderMoney(mainOrder.getOrderMoney());
                                out.collect(dwdOrder);
                            }
                        }

                    }
                }).print();*/

        env.execute();
    }
}
